package org.dei.perla.aggregates;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;

import org.dei.perla.parser.expressions.ConstantTimestamp;
import org.dei.perla.sys.query.llqruntime.datastructures.Record;
import org.dei.perla.utils.pipe.Pipe;
import org.dei.perla.utils.waiter.Waitable;

public class Buffer {
	
	private final Waitable<Buffer> pWaitable;
	private ArrayList<Record> pRecordSet;
	
	private final int pBufferMaxLenght =0;
	/**
	 * Costruisce l'oggetto Buffer, inizializzando il buffer interno con 0 record
	 */
	public Buffer(){
		this(0);
	}
	
	/**
	 * Costruisce l'oggetto Buffer, inizializzando il buffer interno con spazio per tSize record
	 * @param tSize Dimensione iniziale del buffer
	 */
	public Buffer(int tSize) {
		pWaitable = new Waitable<Buffer>(this);
		pRecordSet = new ArrayList<Record>(tSize);
	}
	
	/**
	 * Costruisce l'oggetto Buffer, inizializzando il buffer interno con un set di Record
	 * @param parRecordSet Set di record
	 */
	public Buffer(ArrayList<Record> parRecordSet){
		pRecordSet = new ArrayList<Record>(parRecordSet);
		pWaitable = new Waitable<Buffer>(this);
	}

	

	
	/**
	 * Aggiunge un record al buffer
	 * @param parRecord
	 * @return true se l'oggetto è stato aggiungto correttamente, false altrimenti
	 */
	public boolean addRecord (Record parRecord){
		boolean boolReturn = false; 
		synchronized (pRecordSet) {
			boolReturn = pRecordSet.add(parRecord);
			pWaitable.notifyWaiter();
		}
		return boolReturn;
	}
	
	/**
	 * Elimina un record dal buffer
	 * @param parRecord Il record da eliminare
	 * @return true se il record è stato eliminato correttamente, false altrimenti
	 */
	public boolean delRecord (Record parRecord){
		boolean boolReturn = false; 
		synchronized (pRecordSet) {
			boolReturn = pRecordSet.remove(parRecord);
			pWaitable.notifyWaiter();
		}
		return boolReturn;
	}
	
	/**
	 *  Ritorna l'indice del record avente il timestamp specificato
	 *  ritorna -1 se il record non è stato trovato
	 */
	private int getIndexByTimestamp(Timestamp parTimestamp){
		
		Record tRecord;
	    synchronized (pRecordSet) {
			Iterator<Record> tIteratorRecordSet = pRecordSet.iterator();
		    //Iterator<Constant> tIteratorConstant;
		    ConstantTimestamp tFieldTimestamp;
		    //Record tRecord;
		    
			while(tIteratorRecordSet.hasNext()){
				tRecord = (Record)tIteratorRecordSet.next();
				
				tFieldTimestamp = tRecord.getTimestamp();
				
				if(tFieldTimestamp.getValueTimestamp() == parTimestamp.getTime()){
					pWaitable.notifyWaiter();
					return pRecordSet.indexOf(tRecord);
				}
			}
			pWaitable.notifyWaiter();
		}
		return -1;
	}
	
	/**
	 * Ritorna l'iteratore di una finestra di record  
	 * @param parInit indice iniziale della finestra
	 * @param parEnd  indice finale della finestra
	 * @return Iterator
	 */
	private Iterator <Record> recordsWindowIteratorByIndexes(int parInit, int parEnd){
		ArrayList <Record> tRecordWindowIterator;
		synchronized (pRecordSet) {
			if (parInit>=0 && parEnd < pRecordSet.size() && parInit <= parEnd ){
				tRecordWindowIterator = new ArrayList<Record>(pRecordSet.subList(parInit, parEnd));
				pWaitable.notifyWaiter();
				return tRecordWindowIterator.iterator();
			}
			else{
				pWaitable.notifyWaiter();
				return null;
			}
		}
	}
	
	/**
	 *  Ritorna il record di indice specificato
	 */
	private Record recordByIndex(int parIndex){
		synchronized (pRecordSet) {
			if (parIndex >=0 && parIndex < pRecordSet.size()){
				pWaitable.notifyAll();
				return (Record)pRecordSet.get(parIndex);
			}
			else{
				pWaitable.notifyAll();
				return null;
			}
						
		}

	}
	
	public Iterator<Record> getIterator(Timestamp parTimestamp, long parDelta){
		int tIndexInit = getIndexByTimestamp(parTimestamp);
		int tIndexEnd  = getIndexByTimestamp(new Timestamp(parTimestamp.getTime() + parDelta));
		
		return recordsWindowIteratorByIndexes ( tIndexInit, tIndexEnd+1);
	}
	
	public Iterator<Record> getIterator (Timestamp parTimestamp, int parNumRecord){
		int tIndexInit = getIndexByTimestamp(parTimestamp);
		
		return recordsWindowIteratorByIndexes(tIndexInit, tIndexInit + parNumRecord);
		
	}
	
	/**
	 * Ritorna il contenuto del buffer
	 */
	public ArrayList<Record> getBuffer() {
		//TODO, take into consideration that here is returning the object itself and not a copy of the 
		//object, even if is synchronized.
		return this.pRecordSet;
	}
	/**
	 * aca
	 * Restituisce un riferimento all'oggetto Waitable
	 */	
	public Waitable<Buffer> getWaitable() {	
		return pWaitable;
	}

	/**
	 * Restituisce la stringa che raccoglie il contenuto del buffer
	 */
	public String toString(){
		
		String tString = "";
		
		synchronized (pRecordSet) {
			Iterator<Record> tIterator = pRecordSet.iterator();
			while(tIterator.hasNext())
				tString = tString + ((Record)tIterator.next()).toString() + "\n";
			pWaitable.notifyAll();
			return tString;
		}

	}
	
}
